from faster_whisper import WhisperModel
import yt_dlp
import os
import tempfile
import requests
from moviepy.editor import VideoFileClip
from urllib.parse import urlparse
import json
import asyncio
import aiohttp
from pathlib import Path
from typing import List, Dict, Optional
import time


class ModernVideoTextExtractor:
    def __init__(self, model_size: str = "large-v3", device: str = "auto", compute_type: str = "auto"):
        """
        使用最新的faster-whisper技术栈

        Args:
            model_size: tiny, base, small, medium, large-v2, large-v3 (最新)
            device: auto, cpu, cuda
            compute_type: auto, int8, int16, float16, float32
        """
        print(f"🚀 初始化最新Whisper模型: {model_size}")

        # 使用最新的faster-whisper，性能比原版提升5-10倍
        self.model = WhisperModel(
            model_size,
            device=device,
            compute_type=compute_type,
            cpu_threads=os.cpu_count(),  # 利用所有CPU核心
            num_workers=4  # 并行处理
        )
        print("✅ 最新模型加载完成")

    async def download_video_async(self, url: str, output_path: Optional[str] = None) -> str:
        """异步下载视频，使用最新的并发技术"""
        if output_path is None:
            output_path = tempfile.mkdtemp()

        # 判断是否为流媒体URL
        streaming_domains = [
            'youtube.com', 'youtu.be', 'bilibili.com', 'tiktok.com',
            'instagram.com', 'twitter.com', 'x.com', 'douyin.com'
        ]

        if any(domain in url.lower() for domain in streaming_domains):
            return await self._download_with_ytdlp_async(url, output_path)
        else:
            return await self._download_direct_async(url, output_path)

    async def _download_with_ytdlp_async(self, url: str, output_path: str) -> str:
        """使用最新yt-dlp异步下载"""
        output_template = os.path.join(output_path, 'video.%(ext)s')

        # 使用最新的yt-dlp配置
        ydl_opts = {
            'outtmpl': output_template,
            'format': 'best[height<=1080]/best[height<=720]/best',  # 支持更高清晰度
            'noplaylist': True,
            'extract_flat': False,
            'writesubtitles': False,
            'writeautomaticsub': False,
            'concurrent_fragments': 8,  # 并发下载片段
        }

        def download_sync():
            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
                ydl.download([url])
                # 查找下载的文件
                for file in os.listdir(output_path):
                    if file.startswith('video.'):
                        return os.path.join(output_path, file)
                raise FileNotFoundError("下载的视频文件未找到")

        # 在线程池中异步执行
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(None, download_sync)

    async def _download_direct_async(self, url: str, output_path: str) -> str:
        """使用最新异步HTTP客户端下载"""
        timeout = aiohttp.ClientTimeout(total=300)  # 5分钟超时

        async with aiohttp.ClientSession(timeout=timeout) as session:
            print("🌐 开始异步下载...")

            async with session.get(url) as response:
                response.raise_for_status()

                filename = self._get_filename_from_url(url, response.headers)
                filepath = os.path.join(output_path, filename)

                total_size = int(response.headers.get('content-length', 0))
                downloaded_size = 0

                with open(filepath, 'wb') as f:
                    async for chunk in response.content.iter_chunked(8192):
                        f.write(chunk)
                        downloaded_size += len(chunk)

                        if total_size > 0:
                            progress = (downloaded_size / total_size) * 100
                            print(f"\r📥 下载进度: {progress:.1f}%", end='', flush=True)

                print(f"\n✅ 下载完成: {filepath}")
                return filepath

    def _get_filename_from_url(self, url: str, headers) -> str:
        """从URL或响应头获取文件名"""
        # 尝试从Content-Disposition获取
        content_disposition = headers.get('Content-Disposition', '')
        if content_disposition:
            import re
            filename_match = re.findall(r'filename[^;=\n]*=(([\'"]).*?\2|[^;\n]*)', content_disposition)
            if filename_match:
                filename = filename_match[0][0].strip('"\'')
                if filename:
                    return filename

        # 从URL路径获取
        parsed_url = urlparse(url)
        filename = os.path.basename(parsed_url.path)
        if not filename or '.' not in filename:
            filename = f'video_{int(time.time())}.mp4'

        return filename

    def extract_audio_segment_modern(self, video_path: str, start_ms: int, end_ms: int,
                                     output_path: Optional[str] = None) -> str:
        """使用最新技术提取音频片段"""
        if output_path is None:
            output_path = tempfile.mktemp(suffix='.wav')

        try:
            print(f"🎵 使用最新MoviePy处理音频...")

            # 使用最新的VideoFileClip配置
            video = VideoFileClip(video_path, audio_fps=16000)  # 优化音频采样率

            if not video.audio:
                raise ValueError("❌ 视频文件没有音频轨道")

            audio = video.audio
            start_sec = start_ms / 1000.0
            end_sec = end_ms / 1000.0

            print(f"📊 视频信息: 总时长 {video.duration:.2f}秒")
            print(f"⏱️  提取时间段: {start_sec:.2f}s - {end_sec:.2f}s")

            # 时间范围检查和调整
            if start_sec >= video.duration:
                raise ValueError(f"❌ 开始时间 {start_sec}s 超出视频长度 {video.duration}s")
            if end_sec > video.duration:
                end_sec = video.duration
                print(f"⚠️  结束时间调整为视频结尾: {end_sec}s")

            # 提取音频片段
            audio_segment = audio.subclip(start_sec, end_sec)

            # 使用最新的音频导出配置
            audio_segment.write_audiofile(
                output_path,
                verbose=False,
                logger=None,
                codec='pcm_s16le',  # 高质量音频编码
                ffmpeg_params=['-ar', '16000']  # 优化采样率
            )

            audio.close()
            video.close()

            print(f"✅ 音频提取完成: {output_path}")
            return output_path

        except Exception as e:
            print(f"❌ 音频提取失败: {e}")
            raise

    def transcribe_with_latest_whisper(self, audio_path: str, language: Optional[str] = None) -> Dict:
        """使用最新的faster-whisper进行转录"""
        try:
            print("🎤 开始最新AI语音识别...")

            # 使用最新的faster-whisper API
            segments, info = self.model.transcribe(
                audio_path,
                language=language,
                beam_size=5,  # 更好的准确性
                word_timestamps=True,  # 单词级时间戳
                vad_filter=True,  # 语音活动检测
                vad_parameters=dict(
                    min_silence_duration_ms=500,
                    threshold=0.5
                )
            )

            print(f"🌍 检测语言: {info.language} (置信度: {info.language_probability:.2f})")
            print(f"⚡ 处理时长: {info.duration:.2f}秒")

            # 转换为标准格式
            result_segments = []
            for segment in segments:
                segment_data = {
                    'start': segment.start,
                    'end': segment.end,
                    'text': segment.text,
                    'words': []
                }

                if hasattr(segment, 'words') and segment.words:
                    for word in segment.words:
                        segment_data['words'].append({
                            'word': word.word,
                            'start': word.start,
                            'end': word.end,
                            'probability': word.probability
                        })

                result_segments.append(segment_data)

            return {
                'segments': result_segments,
                'language': info.language,
                'language_probability': info.language_probability,
                'duration': info.duration
            }

        except Exception as e:
            print(f"❌ 语音识别失败: {e}")
            raise

    async def get_text_in_timerange_async(self, url: str, start_ms: int, end_ms: int) -> Dict:
        """异步获取视频指定时间段的文本"""
        video_path = None
        audio_path = None

        try:
            print(f"🚀 开始处理时间段: {start_ms}ms - {end_ms}ms")

            # 1. 异步下载视频
            print(f"📺 下载视频: {url}")
            video_path = await self.download_video_async(url)
            print(f"✅ 视频下载完成")

            # 2. 提取音频片段
            audio_path = self.extract_audio_segment_modern(video_path, start_ms, end_ms)

            # 3. 最新AI语音识别
            result = self.transcribe_with_latest_whisper(audio_path)

            # 4. 处理结果，添加时间偏移
            text_segments = []
            for segment in result['segments']:
                segment_info = {
                    'start_ms': int((segment['start'] * 1000) + start_ms),
                    'end_ms': int((segment['end'] * 1000) + start_ms),
                    'text': segment['text'].strip(),
                    'words': []
                }

                # 处理单词级时间戳
                for word in segment.get('words', []):
                    word_info = {
                        'word': word['word'],
                        'start_ms': int((word['start'] * 1000) + start_ms),
                        'end_ms': int((word['end'] * 1000) + start_ms),
                        'probability': word['probability']
                    }
                    segment_info['words'].append(word_info)

                text_segments.append(segment_info)

            return {
                'success': True,
                'segments': text_segments,
                'full_text': ' '.join([seg['text'] for seg in text_segments]),
                'time_range': f"{start_ms}ms - {end_ms}ms",
                'language': result['language'],
                'language_probability': result['language_probability'],
                'processing_duration': result['duration']
            }

        except Exception as e:
            import traceback
            error_detail = traceback.format_exc()
            print(f"❌ 处理失败: {e}")

            return {
                'success': False,
                'error': str(e),
                'error_detail': error_detail,
                'time_range': f"{start_ms}ms - {end_ms}ms"
            }

        finally:
            # 清理临时文件
            await self._cleanup_files_async([video_path, audio_path])

    async def _cleanup_files_async(self, file_paths: List[str]):
        """异步清理临时文件"""

        def cleanup_sync():
            for file_path in file_paths:
                if file_path and os.path.exists(file_path):
                    try:
                        os.remove(file_path)
                        print(f"🗑️  已清理: {os.path.basename(file_path)}")
                    except Exception as e:
                        print(f"⚠️  清理失败 {file_path}: {e}")

        loop = asyncio.get_event_loop()
        await loop.run_in_executor(None, cleanup_sync)


# 使用最新技术的主函数
async def main_modern():
    """使用最新技术栈的主函数"""
    print("🚀 启动最新视频文本分析系统")
    print("=" * 60)

    # 使用最新的large-v3模型
    extractor = ModernVideoTextExtractor(
        model_size="large-v3",  # 最新最强模型
        device="auto",  # 自动选择最佳设备
        compute_type="auto"  # 自动选择最佳计算类型
    )

    # 您的视频URL
    video_url = "https://mowen-projects.oss-cn-beijing.aliyuncs.com/sound-small-program/resource/ThreeGradeUp/situation/3001/video/GradeThreeUp_U1_P12.mp4"

    # 获取特定时间段的文本
    start_time_ms = 10500  # 10.5秒
    end_time_ms = 25800  # 25.8秒

    result = await extractor.get_text_in_timerange_async(video_url, start_time_ms, end_time_ms)

    if result['success']:
        print(f"\n🎉 分析成功!")
        print(f"📊 时间段: {result['time_range']}")
        print(f"🌍 检测语言: {result['language']} (置信度: {result.get('language_probability', 0):.2f})")
        print(f"⚡ 处理耗时: {result.get('processing_duration', 0):.2f}秒")
        print(f"📝 完整文本: {result['full_text']}")

        print(f"\n📋 详细片段:")
        for i, segment in enumerate(result['segments'], 1):
            print(f"  🎯 片段 {i}: {segment['start_ms']}ms - {segment['end_ms']}ms")
            print(f"  💬 文本: {segment['text']}")

            if segment['words']:
                print(f"  🔤 单词详情:")
                for word in segment['words']:
                    print(
                        f"    '{word['word']}' ({word['start_ms']}ms-{word['end_ms']}ms, 置信度:{word['probability']:.3f})")
            print()
    else:
        print(f"❌ 分析失败: {result['error']}")


# 同步版本（如果不需要异步）
def main_sync():
    """同步版本"""
    asyncio.run(main_modern())


if __name__ == "__main__":
    # 使用最新的异步技术
    asyncio.run(main_modern())

    # 或者使用同步版本
    # main_sync()
